package org.qiyu.live.user.provider.config;

import com.alibaba.fastjson2.JSON;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.util.internal.ObjectUtil;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.qiyu.live.framework.redis.starter.key.UserProviderCacheKeyBuilder;
import org.qiyu.live.user.constants.CacheAsyncDeleteCode;
import org.qiyu.live.user.dto.UserCacheAsyncDeleteDTO;
import org.qiyu.live.user.dto.UserDTO;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static org.qiyu.live.user.constants.UserProviderTopicNames.CACHE_ASYNC_DELETE_TOPIC;

/**
 * @Author 朝问道
 * @Date 2024年01月11日 9:37
 * @Description 消费者的配置信息
 */

//@Configuration
@Slf4j
@Configuration
public class RocketMQConsumerConfig implements InitializingBean{


    @Resource
    private RocketMQConsumerProperties rocketMQConsumerProperties;
    @Resource
    private ObjectMapper objectMapper;

    @Resource
    private RedisTemplate redisTemplate;

    @Resource
    private UserProviderCacheKeyBuilder userProviderCacheKeyBuilder;

    @Resource
    private UserProviderCacheKeyBuilder cacheKeyBuilder;



    @Override
    public void afterPropertiesSet() throws Exception {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
        defaultMQPushConsumer.setNamesrvAddr(rocketMQConsumerProperties.getNameServer());
        defaultMQPushConsumer.setConsumerGroup(rocketMQConsumerProperties.getGroupName() + "_" + RocketMQConsumerConfig.class.getSimpleName());
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(1); //一次只拉一条消息
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        try {
            defaultMQPushConsumer.subscribe(CACHE_ASYNC_DELETE_TOPIC, "*");
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
        defaultMQPushConsumer.setMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            String msgStr = new String(list.get(0).getBody());

            try {
                UserCacheAsyncDeleteDTO userCacheAsyncDeleteDTO = objectMapper.readValue(msgStr, UserCacheAsyncDeleteDTO.class);
                if (userCacheAsyncDeleteDTO.getCode() == CacheAsyncDeleteCode.USER_INFO_DELETE.getCode()) {
                    //用户基础信息删除
                    Long userId = JSON.parseObject(userCacheAsyncDeleteDTO.getJson()).getLong("userId");
                    redisTemplate.delete(cacheKeyBuilder.buildUserInfoKey(userId));
                    log.info("延迟删除用户信息缓存，userId is {}",userId);
                }
                else if (userCacheAsyncDeleteDTO.getCode() == CacheAsyncDeleteCode.USER_TAG_DELETE.getCode()) {
                    //用户标签信息删除
                    Long userId = com.alibaba.fastjson.JSON.parseObject(userCacheAsyncDeleteDTO.getJson()).getLong("userId");
                    redisTemplate.delete(cacheKeyBuilder.buildTagKey(userId));
                    log.info("延迟删除用户标签缓存，userId is {}",userId);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;


            } catch (JsonProcessingException e) {
                throw new RuntimeException(e);
            }


//            UserDTO userDTO = null;
//            try {
//                userDTO = objectMapper.readValue(msgStr, UserDTO.class);
//                // 等价于 JSON.parseObject(msgStr, UserDTO.class); 作用都是将json格式的字符串转化为对象实例
//                if (userDTO == null || userDTO.getUserId() == null) {
//                    log.error("用户id为空,参数异常,内容:{}", msgStr);
//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                }
//                redisTemplate.delete(userProviderCacheKeyBuilder.buildUserInfoKey(userDTO.getUserId()));
//                log.info("延迟删除处理,userID is {}", userDTO.getUserId());
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//
//
//            } catch (JsonProcessingException e) {
//                throw new RuntimeException(e);
//            }


        });

        defaultMQPushConsumer.start();
        log.info("MQ消费者启动成功, name server: {}", rocketMQConsumerProperties.getNameServer());
    }
}
